package com.hzgj.bcl.soa.rpc.server;

import com.hzgj.bcl.soa.Event;
import com.hzgj.bcl.soa.cicada.pool.Channels;
import com.hzgj.bcl.soa.rpc.client.Request;
import com.hzgj.bcl.soa.rpc.exception.RpcException;
import com.hzgj.bcl.soa.rpc.exception.RpcInfo;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.util.lang.StrKit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.AllArgsConstructor;

public class NettyServerHandler extends SimpleChannelInboundHandler<Message> {

    private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);
    private static final ChannelFutureListener listener = future -> {
        if (!future.isSuccess()) {
            logger.error("tp write response error");
        }
    };
    private ExecutorService threadPool;
    private NettyServer nettyServer;
    private String serverHost;
    private int serverPort;

    public NettyServerHandler(NettyServer nettyServer, ExecutorService threadPool) {
        this.threadPool = threadPool;
        this.nettyServer = nettyServer;
        serverHost = nettyServer.getServerConfig().getHost();
        if (StrKit.isBlank(serverHost)) {
            try {
                serverHost = InetAddress.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                logger.error("{}", e);
            }
        }
        serverPort = nettyServer.getServerConfig().getPort();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable e)
            throws Exception {
        if (!(e instanceof IOException)) {
            logger.error("catch some exception not IOException", e);
        }
        Channels.setDiscard(ctx.channel());
        Channels.silentlyCloseChannel(ctx.channel());
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
        try {
            threadPool.execute(new HandlerRunnable(ctx.channel(), msg));
        } catch (RejectedExecutionException exception) {
            logger.error("biz threadPool full,threadPool maxsize is:" + ((ThreadPoolExecutor) threadPool).getMaximumPoolSize());
            Message result = new Message();
            result.putHeader(Constants.MESSAGE_SID, msg.getHeader(Constants.MESSAGE_SID))
                    .setPayload(new RpcException(RpcInfo.SERVER_BUSY));
            sendErrorResponse(ctx.channel(), result);
        }
    }

    private void sendErrorResponse(final Channel channel, final Message message) {
        channel.writeAndFlush(message).
                addListener((ChannelFutureListener) future -> {
                    if (!future.isSuccess()) {
                        logger.error("biz write response error, message is: " + message);
                    }
                });
    }

    @AllArgsConstructor
    class HandlerRunnable implements Runnable {

        private Channel channel;
        private Message message;

        @Override
        public void run() {
            Long req_id = (Long) message.getHeader(Constants.MESSAGE_SID);
            if (req_id == null) {
                //针对协议中不包含SID的情况
                req_id = ThreadLocalRandom.current().nextLong() + Request.newId();
                message.putHeader(Constants.MESSAGE_SID, req_id);
            }
            Channels.addChannel(req_id, channel);
            InetSocketAddress address = (InetSocketAddress) channel.remoteAddress();
            message.putHeader(Constants.MESSAGE_HEADER_REMOTE_HOST, address.getAddress().getHostAddress())
                    .putHeader(Constants.MESSAGE_HEADER_REMOTE_PORT, address.getPort())
                    .putHeader(Constants.MESSAGE_HEADER_SERVER_HOST, serverHost)
                    .putHeader(Constants.MESSAGE_HEADER_SERVER_PORT, serverPort);
            nettyServer.up(new Event(Event.MSG, message));
        }
    }
}
